package com.gjy.learning.scala.sql

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{add_months, current_date, desc, lit, months_between, row_number, sum, to_date}

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

/** *
 * 表1：
 * 人id 账号id
 *
 * 表2：
 * 账号id ，金额，进出标记，日期
 *
 * 输出金额从今年往前数12个月最大的3个账号
 */
object TableJoinAndDateFillter {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("TaskOne").master("local[*]").getOrCreate()

    val data1 = Seq(
      ("u1", "c1"),
      ("u1", "c2"),
      ("u2", "c5"),
      ("u3", "c6"),
      ("u4", "c7"),
      ("u5", "c8"),
      ("u6", "c9")
    )
    val schema1 = List(
      "custmer_id", "account_number"
    )

    // Define the schema
    val data2 = Seq(
      ("c1", "1234", "-", "2023-02-11"),
      ("c2", "452", "O", "2023-12-20"),
      ("c3", "3543", "-", "2024-02-11"),
      ("c4", "41", "O", "2022-02-11"),
      ("c5", "511", "O", "2024-02-11"),
      ("c6", "622", "-", "2022-02-11"),
      ("c7", "744", "-", "2022-02-11"),
      ("c8", "833", "O", "2024-02-11"),
      ("c9", "911", "O", "2024-02-11"),
      ("c1", "110", "-", "2024-02-11"),
      ("c2", "141", "-", "2024-02-11")
    )
    val schema2 = List("account_number", "transaction_amount", "in_or_out", "tran_date")




/*
    val currentTime = LocalDateTime.now()
    println(currentTime)
    println(currentTime.minusMonths(12))
    val twelveMonthsAgo = currentTime.minusMonths(12).format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))
    val currentDate = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))
*/

    import spark.implicits._
    import com.gjy.learning.scala.util.GetUtil.getDFScheamaType
    // Create the DataFrame
    val df1 = data1.toDF(schema1: _*)

    val df2 = data2.toDF(schema2: _*)

    df2.filter(months_between(lit(current_date()),to_date($"tran_date", "yyyy-MM-dd")) <= 12).show()


    val currentDate = current_date() // 获取当前日期
    val twelveMonthsAgo = add_months(currentDate, -12) // 计算12个月前的日期

    df2.filter(to_date($"tran_date", "yyyy-MM-dd") >= twelveMonthsAgo)
      .filter(to_date($"tran_date", "yyyy-MM-dd") <= currentDate)
      .show()
/*

    val dfWithDatefiler = df2.filter($"tran_date" >= twelveMonthsAgo)
      .filter($"tran_date" <= currentDate)
      .select("account_number", "transaction_amount")


    val ordernum = 3
    dfWithDatefiler.show()
    val result = df1.join(dfWithDatefiler,
        df1.col("account_number") === dfWithDatefiler.col("account_number"))
      .select("custmer_id", "transaction_amount")
      .groupBy("custmer_id")
      .agg(sum("transaction_amount").as("total_amount"))
      .withColumn("rn", row_number().over(Window.orderBy(desc("total_amount"))))
      .filter($"rn" <= ordernum)
      .sort("rn")

    result.show()
*/

  spark.stop()
  }

  /*
    df1.join(df2, col("account_number"), "inner")
      .select("custmer_id", "account_number")
      .groupBy("custmer_id")
      .agg(sum("amount").as("total_amount"))*/
}
